package com.hadoop.gmall.realtime.dwd.db.app;

import com.bw.gmall.realtime.common.base.BaseSqlApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeRefundPaySucDetail extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeRefundPaySucDetail().start(Constant.TOPIC_DB, Constant.TOPIC_DWD_TRADE_REFUND_PAYMENT_SUCCESS, 4, 10018);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv, String groupId) {
        // 1. 读取 topic_db
        readOdsDb(tEnv, Constant.TOPIC_DWD_TRADE_REFUND_PAYMENT_SUCCESS);
        // 2. 读取 字典表
        readBaseDic(tEnv);

        // 3. 过滤退款成功表数据
        Table refundPayment = tEnv.sqlQuery(
                "SELECT " +
                        "data['id'] id, " +
                        "data['order_id'] order_id, " +
                        "data['sku_id'] sku_id, " +
                        "data['payment_type'] payment_type, " +
                        "data['callback_time'] callback_time, " +
                        "data['total_amount'] total_amount, " +
                        "proc_time, " +
                        "ts " +
                        "FROM topic_db " +
                        "WHERE `database` = 'gmall' " +
                        "AND `table` = 'refund_payment' " +
                        "AND `type` = 'update' " +
                        "AND `old`['refund_status'] IS NOT NULL " +
                        "AND `data`['refund_status'] = '1602'");
        tEnv.createTemporaryView("refund_payment", refundPayment);

        // 4. 过滤退单表中的退单成功的数据
        Table orderRefundInfo = tEnv.sqlQuery(
                "SELECT " +
                        "data['order_id'] order_id, " +
                        "data['sku_id'] sku_id, " +
                        "data['refund_num'] refund_num " +
                        "FROM topic_db " +
                        "WHERE `database` = 'gmall' " +
                        "AND `table` = 'order_refund_info' " +
                        "AND `type` = 'update' " +
                        "AND `old`['refund_status'] IS NOT NULL " +
                        "AND `data`['refund_status'] = '0705'");
        tEnv.createTemporaryView("order_refund_info", orderRefundInfo);

        // 5. 过滤订单表中的退款成功的数据
        Table orderInfo = tEnv.sqlQuery(
                "SELECT " +
                        "data['id'] id, " +
                        "data['user_id'] user_id, " +
                        "data['province_id'] province_id " +
                        "FROM topic_db " +
                        "WHERE `database` = 'gmall' " +
                        "AND `table` = 'order_info' " +
                        "AND `type` = 'update' " +
                        "AND `old`['order_status'] IS NOT NULL " +
                        "AND `data`['order_status'] = '1006'");
        tEnv.createTemporaryView("order_info", orderInfo);

        // 6. 4 张表的 join
        Table result = tEnv.sqlQuery(
                "SELECT " +
                        "rp.id, " +
                        "oi.user_id, " +
                        "rp.order_id, " +
                        "rp.sku_id, " +
                        "oi.province_id, " +
                        "rp.payment_type, " +
                        "dic.dic_name payment_type_name, " +
                        "DATE_FORMAT(rp.callback_time, 'yyyy-MM-dd') date_id, " +
                        "rp.callback_time, " +
                        "ori.refund_num, " +
                        "rp.total_amount, " +
                        "rp.ts " +
                        "FROM refund_payment rp " +
                        "JOIN order_refund_info ori " +
                        "ON rp.order_id = ori.order_id AND rp.sku_id = ori.sku_id " +
                        "JOIN order_info oi " +
                        "ON rp.order_id = oi.id " +
                        "JOIN base_dic FOR SYSTEM_TIME AS OF rp.proc_time AS dic " +
                        "ON rp.payment_type = dic.dic_code ");

        // 7.写出到 kafka
        tEnv.executeSql("CREATE TABLE dwd_trade_refund_payment_success (" +
                "id STRING, " +
                "user_id STRING, " +
                "order_id STRING, " +
                "sku_id STRING, " +
                "province_id STRING, " +
                "payment_type_code STRING, " +
                "payment_type_name STRING, " +
                "date_id STRING, " +
                "callback_time STRING, " +
                "refund_num STRING, " +
                "refund_amount STRING, " +
                "ts BIGINT " +
                ")" + SQLUtil.getKafkaDDLSink(Constant.TOPIC_DWD_TRADE_REFUND_PAYMENT_SUCCESS));
        result.executeInsert(Constant.TOPIC_DWD_TRADE_REFUND_PAYMENT_SUCCESS);
    }
}